package cn.jly.flink.table;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;

/**
 * flink老、新版本的table api & SQL的执行入口的创建
 *
 * @author lanyangji
 * @date 2020/7/22
 */
public class FlinkTableApiBeginningApp {
    public static void main(String[] args) {
        // ====================================flink table api & SQL===================================================
        // flinke stream table query
        EnvironmentSettings streamEnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment, streamEnvironmentSettings);

        // flink batch table query
        ExecutionEnvironment batchExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        // 不需要传入配置类，默认就是创建批处理tableEnv
        BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(batchExecutionEnvironment);



        // ====================================blink table api & SQL===================================================
        // blink stream table query
        StreamExecutionEnvironment blinkStreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings blinkStreamEnvironmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment blinkStreamTabEnv = StreamTableEnvironment.create(blinkStreamExecutionEnvironment, blinkStreamEnvironmentSettings);

        // blink batch table query
        EnvironmentSettings blinkBatchEnvironmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(blinkBatchEnvironmentSettings);
    }
}
